package com.xiaomaoguai;

import com.alibaba.fastjson.JSONObject;
import com.zhongan.blcs.client.BlcsAssit;
import com.zhongan.blcs.client.DBEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.io.IOException;

/**
 * @author WeiHui-Z
 * @version v1.0.0
 * @date 2019/8/26 14:09
 * @since JDK 1.8
 */
@Slf4j
public class DecisionApplyDtoDeserializationSchema implements DeserializationSchema<DBEvent> {

	@Override
	public DBEvent deserialize(byte[] bytes) throws IOException {
		DBEvent dbEvent = BlcsAssit.parse(bytes);
		String dbEventJsonString = JSONObject.toJSONString(dbEvent);
		log.info("DBEvent:{}", dbEventJsonString);
		return dbEvent;
	}

	@Override
	public boolean isEndOfStream(DBEvent dbEvent) {
		return false;
	}

	@Override
	public TypeInformation<DBEvent> getProducedType() {
		return TypeInformation.of(DBEvent.class);
	}

}
